package cn.lcc.utils.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * ffmpeg -i '源文件.mp4' -c:v h264 -flags +cgop -g 30 -hls_time 20 -hls_list_size 0 -hls_segment_filename index%3d.ts 'index.m3u8'
 */
@Slf4j
@Service
public class BaseFluxServiceImpl implements BaseFluxService {

    private void createData(FluxSink<String> sink) {
        Flux.interval(Duration.ofSeconds(5)).map(i -> {
                    log.info("发布数据");
                    return "Flux data---" + i;
                })
                .doOnNext(sink::next)
                .subscribe();
    }

    public static Flux<String> source;
    public static Flux<String> result;

    private final Map<String, MonoSink> topicSubscribers = new HashMap<>();

    @Override
    public Mono generateData2() {
        log.info("创建监听");
        return Mono.create(sink -> topicSubscribers.put("test", sink));
    }

    @Override
    public void unsubscribe() {
        log.info("卸载监听");
        topicSubscribers.remove("test");
    }

    @Async
    @Override
    public Flux publishMsg(String msg) {
            MonoSink<String> mono = topicSubscribers.get("test");
            if (mono != null) {
                log.info("检测到监听者,发送消息");
                mono.success(msg);
            }
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return null;
    }



}



















